package HignestTemperature;

import java.io.IOException;



import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.fs.Path;

import org.apache.hadoop.io.IntWritable;

import org.apache.hadoop.io.LongWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Job;

import org.apache.hadoop.mapreduce.Mapper;

import org.apache.hadoop.mapreduce.Reducer;

import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

/**
 * Created by luoqifei on 2016/11/1.
 */
public class Temperature {
    //the Mapper class para <LongWritable, Text。。is read in,and Text, IntWritable> is output
    static class TempMapper extends Mapper<LongWritable, Text, Text, IntWritable>{
        @Override
        //this map method,(LongWritable key,Text value is in, and context is output
        public void map(LongWritable key,Text value,Context context) throws IOException, InterruptedException {
            System.out.print("Before Mapper: " + key + ", " + value);
            String line = value.toString();
            String year = line.substring(0,4);
            int temperature = Integer.parseInt(line.substring(8));
            context.write(new Text(year),new IntWritable(temperature));
            System.out.println("======" +"After Mapper:" + new Text(year) + ", " + new IntWritable(temperature));
        }
        }
    static class TempReducer extends Reducer<Text,IntWritable,Text,IntWritable>{
        @Override
        //key and values is map output,so as reducer input
        public void reduce(Text key,Iterable<IntWritable> values,Context context)throws IOException,InterruptedException{
            int maxValue = Integer.MIN_VALUE;
            StringBuilder sb = new StringBuilder();
            for(IntWritable value : values){
                    maxValue = Math.max(maxValue,value.get());
                sb.append(value).append(", ");
            }
            System.out.print("Before Reduce: " + key + ", " + sb.toString());
            context.write(key,new IntWritable(maxValue));
            System.out.println("======" + "After Reduce: " + key + ", " + maxValue);
        }
    }
    public static void main(String[] args) throws Exception {
        //set hdfs url
        String dst = "hdfs://localhost:9000/user/input.txt";
        //dstOut must be not exist,or will be error
        String dstOut = "hdfs://localhost:9000/temperature1";
        Configuration hadoopConf = new Configuration();
        hadoopConf.set("fs.hdfs.impl",org.apache.hadoop.hdfs.DistributedFileSystem.class.getName());
        //只能是本地模式的file system
        hadoopConf.set("fs.file.impl",org.apache.hadoop.fs.LocalFileSystem.class.getName());


        //init the MR job
        Job job = new Job(hadoopConf);
        //如果需要打成jar运行，需要下面这句
        job.setJarByClass(Temperature.class);

        FileInputFormat.addInputPath(job,new Path(dst));
        FileOutputFormat.setOutputPath(job,new Path(dstOut));
        //set mapper and reducer
        job.setMapperClass(TempMapper.class);
        job.setReducerClass(TempReducer.class);
        //set output type
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        //set wait until finish the job
        job.waitForCompletion(true);
        System.out.println("Finished");
    }
}
